package com.adam.example;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.filecache.DistributedCache;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.elasticsearch.hadoop.mr.EsOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class HdfsToES {

    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsToES.class);

    public static class MyMapper extends Mapper<Object, Text, NullWritable, Text> {
        //Mapper的四个泛型参数分别为KEYIN, VALUEIN, KEYOUT, VALUEOUT
        public void map(Object key, Text value, Mapper<Object, Text, NullWritable, Text>.Context context)
                throws IOException, InterruptedException {
            byte[] line = value.toString().trim().getBytes();
            Text blog = new Text(line);
            context.write(NullWritable.get(), blog);
        }
    }

    //打包成jar在虚拟机中使用hadoop jar命令运行
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        LOGGER.debug("For debug only --- start");
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        URLClassLoader urlClassLoader = (URLClassLoader) classLoader;
        LOGGER.debug("Context ClassLoader class={}", classLoader.getClass());
        Arrays.stream(urlClassLoader.getURLs()).map(URL::getPath).forEach(LOGGER::debug);
        String classpath = System.getProperty("java.class.path");
        LOGGER.debug("classpath: {}", classpath);
        LOGGER.debug("env HADOOP_USE_CLIENT_CLASSLOADER=" + System.getenv("HADOOP_USE_CLIENT_CLASSLOADER"));
        Class<?> esOutputFormatClass = urlClassLoader.loadClass("org.elasticsearch.hadoop.mr.EsOutputFormat");
        LOGGER.debug("load EsOutputFormat 1:{}", esOutputFormatClass);
        Class<?> esOutputFormatClass2 = Class.forName("org.elasticsearch.hadoop.mr.EsOutputFormat");
        LOGGER.debug("load EsOutputFormat 2:{}", esOutputFormatClass2);
        LOGGER.debug("For debug only --- end");

//        System.setProperty("hadoop.home.dir", "D:\\Dependencies\\hadoop\\2.10.2");
        Configuration conf = new Configuration();
        conf.setBoolean("mapred.map.tasks.speculative.execution", false);  //关闭Map阶段执行推测
        conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);
//        conf.set("mapreduce.job.user.classpath.first","true");
//        conf.set("mapreduce.task.classpath.user.precedence","true");
//        conf.set("mapreduce.frameworkclasspath.name", "yarn");
//        conf.set("fs.defaultFS", "hdfs://192.168.0.111:9000/");
        conf.set("es.nodes", "192.168.0.111:9200");
        conf.set("es.resource", "test_blog_2/csdn"); //索引到ES的索引名和类型名
        conf.set("es.mapping.id", "id"); //指定id字段，不指定会随机生成
        conf.set("es.input.json", "yes"); //指定输入文件类型为json
//        DistributedCache.addFileToClassPath(new Path("hdfs://192.168.0.111:9000//library/elasticsearch-hadoop-mr-5.4.0.jar"), conf);  //添加到job执行容器classpath

        Job job = Job.getInstance(conf, "hadoop es write test");
        job.setJarByClass(HdfsToES.class);
        job.setMapperClass(HdfsToES.MyMapper.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(EsOutputFormat.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);
        String osName = System.getProperty("os.name").toLowerCase();
        boolean addESHadoopToClasspath = Boolean.parseBoolean(System.getenv("ADD_ES_HADOOP_TO_CLASSPATH"));
        if(!osName.startsWith("windows") && addESHadoopToClasspath) {
            job.addFileToClassPath(new Path("hdfs://192.168.0.111:9000//library/elasticsearch-hadoop-mr-5.4.0.jar"));
        }

        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.0.111:9000//test_work/blog.json"));
        job.waitForCompletion(true);
    }
}
